// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/bitstamp/feed_v2/processor.h"

#include <utility>

#include <glog/logging.h>
#include <rapidjson/document.h>
#include <rapidjson/error/en.h>

#include "coin2/exchange/bitstamp/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bitstamp::feed_v2::impl {

bool BitstampFeedProcessor::IsKnownMessage(const rapidjson::Value& document) {
  if (!document.HasMember("channel")) {
    return false;
  }

  if (!document.HasMember("event")) {
    return false;
  }

  if (document["event"] == "trade") {  // trade
    return true;
  } else if (document["event"] == "data"){  // book
    return true;
  } else {
    return false;
  }
}

void BitstampFeedProcessor::ParseChennel(
    const std::string& channel,
    std::string* channel_name,
    std::string* native_symbol) {
  // e.g. "order_book_btcusd", "live_trades_btcusd"
  std::size_t underscore = channel.find_last_of('_');
  *channel_name = channel.substr(0, underscore);
  *native_symbol = channel.substr(underscore + 1);
}

void BitstampFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;

  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error("JSON parse error: "s + GetParseError_En(document.GetParseError()));
  }

  if (IsKnownMessage(document)) {
    const std::string& channel = document["channel"].GetString();
    std::string channel_name;
    std::string native_symbol;
    ParseChennel(channel, &channel_name, &native_symbol);
    if (!CheckChannelFilter(channel_name)) {
      return;
    }

    if (channel_name == "order_book") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
          const auto& data =  document["data"];
          bool is_crossed_book = false;
          BitstampFeedUpdateParser::ParseBook(book, data, &is_crossed_book);
          // ignore abnormal feed as the mid price of crossed book is abnormal too
          if (!is_crossed_book) upd_mgr_.PublishBook();
      }
    } else if (channel_name == "live_trades") {
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        const auto& data =  document["data"];
        Trade* trade = upd_mgr_.GetTradeMutable();
        BitstampFeedUpdateParser::ParseTrade(data, trade);
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else {
      Error("[Bitstamp processor] unknown channel:", channel);
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Bitstamp Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }
  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::bitstamp::feed_v2::impl
